{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "(distributed-graph)=\n",
    "# Distributed (multi-function) pipeline example\n",
    "\n",
    "This example demonstrates how to run a pipeline that consists of multiple serverless functions (connected using streams).\n",
    "\n",
    "In the pipeline example the request contains the a URL of a file. It loads the content of the file and breaks it into paragraphs (using the FlatMap class), and pushes the results to a queue/stream. The second function picks up the paragraphs and runs the NLP flow to extract the entities and push the results to the output stream."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Setting the stream URLs for the internal queue, the final output and error/exceptions stream:**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "streams_prefix = \"v3io:///users/admin/\"\n",
    "internal_stream = streams_prefix + \"in-stream\"\n",
    "out_stream = streams_prefix + \"out-stream\"\n",
    "err_stream = streams_prefix + \"err-stream\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Alternatively, using Kafka:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "# kafka_prefix = f\"kafka://{broker}/\"\n",
    "# internal_topic = kafka_prefix + \"in-topic\"\n",
    "# out_topic = kafka_prefix + \"out-topic\"\n",
    "# err_topic = kafka_prefix + \"err-topic\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In either case, continue with:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "> 2025-10-05 10:00:20,453 [info] Project loaded successfully: {\"project_name\":\"pipe\"}\n"
     ]
    }
   ],
   "source": [
    "# set up the environment\n",
    "import mlrun\n",
    "\n",
    "project = mlrun.get_or_create_project(\"pipe\", allow_cross_project=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Collecting spacy\n",
      "  Downloading spacy-3.8.7-cp311-cp311-manylinux_2_17_x86_64.manylinux2014_x86_64.whl.metadata (27 kB)\n",
      "Collecting spacy-legacy<3.1.0,>=3.0.11 (from spacy)\n",
      "  Downloading spacy_legacy-3.0.12-py2.py3-none-any.whl.metadata (2.8 kB)\n",
      "Collecting spacy-loggers<2.0.0,>=1.0.0 (from spacy)\n",
      "  Downloading spacy_loggers-1.0.5-py3-none-any.whl.metadata (23 kB)\n",
      "Collecting murmurhash<1.1.0,>=0.28.0 (from spacy)\n",
      "  Downloading murmurhash-1.0.13-cp311-cp311-manylinux_2_5_x86_64.manylinux1_x86_64.manylinux_2_17_x86_64.manylinux2014_x86_64.whl.metadata (2.2 kB)\n",
      "Collecting cymem<2.1.0,>=2.0.2 (from spacy)\n",
      "  Downloading cymem-2.0.11-cp311-cp311-manylinux_2_17_x86_64.manylinux2014_x86_64.whl.metadata (8.5 kB)\n",
      "Collecting preshed<3.1.0,>=3.0.2 (from spacy)\n",
      "  Downloading preshed-3.0.10-cp311-cp311-manylinux_2_5_x86_64.manylinux1_x86_64.manylinux_2_17_x86_64.manylinux2014_x86_64.whl.metadata (2.4 kB)\n",
      "Collecting thinc<8.4.0,>=8.3.4 (from spacy)\n",
      "  Downloading thinc-8.3.6-cp311-cp311-manylinux_2_17_x86_64.manylinux2014_x86_64.whl.metadata (15 kB)\n",
      "Collecting wasabi<1.2.0,>=0.9.1 (from spacy)\n",
      "  Downloading wasabi-1.1.3-py3-none-any.whl.metadata (28 kB)\n",
      "Collecting srsly<3.0.0,>=2.4.3 (from spacy)\n",
      "  Downloading srsly-2.5.1-cp311-cp311-manylinux_2_17_x86_64.manylinux2014_x86_64.whl.metadata (19 kB)\n",
      "Collecting catalogue<2.1.0,>=2.0.6 (from spacy)\n",
      "  Downloading catalogue-2.0.10-py3-none-any.whl.metadata (14 kB)\n",
      "Collecting weasel<0.5.0,>=0.1.0 (from spacy)\n",
      "  Downloading weasel-0.4.1-py3-none-any.whl.metadata (4.6 kB)\n",
      "Requirement already satisfied: typer<1.0.0,>=0.3.0 in /User/.conda/envs/daniel311/lib/python3.11/site-packages (from spacy) (0.17.3)\n",
      "Requirement already satisfied: tqdm<5.0.0,>=4.38.0 in /User/.conda/envs/daniel311/lib/python3.11/site-packages (from spacy) (4.67.1)\n",
      "Requirement already satisfied: numpy>=1.19.0 in /User/.conda/envs/daniel311/lib/python3.11/site-packages (from spacy) (1.26.4)\n",
      "Requirement already satisfied: requests<3.0.0,>=2.13.0 in /User/.conda/envs/daniel311/lib/python3.11/site-packages (from spacy) (2.32.5)\n",
      "Requirement already satisfied: pydantic!=1.8,!=1.8.1,<3.0.0,>=1.7.4 in /User/.conda/envs/daniel311/lib/python3.11/site-packages (from spacy) (2.10.6)\n",
      "Requirement already satisfied: jinja2 in /User/.conda/envs/daniel311/lib/python3.11/site-packages (from spacy) (3.1.6)\n",
      "Requirement already satisfied: setuptools in /User/.conda/envs/daniel311/lib/python3.11/site-packages (from spacy) (80.9.0)\n",
      "Requirement already satisfied: packaging>=20.0 in /User/.conda/envs/daniel311/lib/python3.11/site-packages (from spacy) (24.2)\n",
      "Collecting langcodes<4.0.0,>=3.2.0 (from spacy)\n",
      "  Downloading langcodes-3.5.0-py3-none-any.whl.metadata (29 kB)\n",
      "Collecting language-data>=1.2 (from langcodes<4.0.0,>=3.2.0->spacy)\n",
      "  Downloading language_data-1.3.0-py3-none-any.whl.metadata (4.3 kB)\n",
      "Requirement already satisfied: annotated-types>=0.6.0 in /User/.conda/envs/daniel311/lib/python3.11/site-packages (from pydantic!=1.8,!=1.8.1,<3.0.0,>=1.7.4->spacy) (0.7.0)\n",
      "Requirement already satisfied: pydantic-core==2.27.2 in /User/.conda/envs/daniel311/lib/python3.11/site-packages (from pydantic!=1.8,!=1.8.1,<3.0.0,>=1.7.4->spacy) (2.27.2)\n",
      "Requirement already satisfied: typing-extensions>=4.12.2 in /User/.conda/envs/daniel311/lib/python3.11/site-packages (from pydantic!=1.8,!=1.8.1,<3.0.0,>=1.7.4->spacy) (4.15.0)\n",
      "Requirement already satisfied: charset_normalizer<4,>=2 in /User/.conda/envs/daniel311/lib/python3.11/site-packages (from requests<3.0.0,>=2.13.0->spacy) (3.4.3)\n",
      "Requirement already satisfied: idna<4,>=2.5 in /User/.conda/envs/daniel311/lib/python3.11/site-packages (from requests<3.0.0,>=2.13.0->spacy) (3.10)\n",
      "Requirement already satisfied: urllib3<3,>=1.21.1 in /User/.conda/envs/daniel311/lib/python3.11/site-packages (from requests<3.0.0,>=2.13.0->spacy) (2.5.0)\n",
      "Requirement already satisfied: certifi>=2017.4.17 in /User/.conda/envs/daniel311/lib/python3.11/site-packages (from requests<3.0.0,>=2.13.0->spacy) (2025.8.3)\n",
      "Collecting blis<1.4.0,>=1.3.0 (from thinc<8.4.0,>=8.3.4->spacy)\n",
      "  Downloading blis-1.3.0-cp311-cp311-manylinux_2_17_x86_64.manylinux2014_x86_64.whl.metadata (7.4 kB)\n",
      "Collecting confection<1.0.0,>=0.0.1 (from thinc<8.4.0,>=8.3.4->spacy)\n",
      "  Downloading confection-0.1.5-py3-none-any.whl.metadata (19 kB)\n",
      "Collecting numpy>=1.19.0 (from spacy)\n",
      "  Downloading numpy-2.3.3-cp311-cp311-manylinux_2_27_x86_64.manylinux_2_28_x86_64.whl.metadata (62 kB)\n",
      "\u001b[2K     \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m62.1/62.1 kB\u001b[0m \u001b[31m2.2 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m\n",
      "\u001b[?25hRequirement already satisfied: click>=8.0.0 in /User/.conda/envs/daniel311/lib/python3.11/site-packages (from typer<1.0.0,>=0.3.0->spacy) (8.2.1)\n",
      "Requirement already satisfied: shellingham>=1.3.0 in /User/.conda/envs/daniel311/lib/python3.11/site-packages (from typer<1.0.0,>=0.3.0->spacy) (1.5.4)\n",
      "Requirement already satisfied: rich>=10.11.0 in /User/.conda/envs/daniel311/lib/python3.11/site-packages (from typer<1.0.0,>=0.3.0->spacy) (14.1.0)\n",
      "Collecting cloudpathlib<1.0.0,>=0.7.0 (from weasel<0.5.0,>=0.1.0->spacy)\n",
      "  Downloading cloudpathlib-0.22.0-py3-none-any.whl.metadata (16 kB)\n",
      "Collecting smart-open<8.0.0,>=5.2.1 (from weasel<0.5.0,>=0.1.0->spacy)\n",
      "  Downloading smart_open-7.3.1-py3-none-any.whl.metadata (24 kB)\n",
      "Requirement already satisfied: MarkupSafe>=2.0 in /User/.conda/envs/daniel311/lib/python3.11/site-packages (from jinja2->spacy) (2.1.5)\n",
      "Collecting marisa-trie>=1.1.0 (from language-data>=1.2->langcodes<4.0.0,>=3.2.0->spacy)\n",
      "  Downloading marisa_trie-1.3.1-cp311-cp311-manylinux_2_24_x86_64.manylinux_2_28_x86_64.whl.metadata (10 kB)\n",
      "Requirement already satisfied: markdown-it-py>=2.2.0 in /User/.conda/envs/daniel311/lib/python3.11/site-packages (from rich>=10.11.0->typer<1.0.0,>=0.3.0->spacy) (4.0.0)\n",
      "Requirement already satisfied: pygments<3.0.0,>=2.13.0 in /User/.conda/envs/daniel311/lib/python3.11/site-packages (from rich>=10.11.0->typer<1.0.0,>=0.3.0->spacy) (2.19.2)\n",
      "Requirement already satisfied: wrapt in /User/.conda/envs/daniel311/lib/python3.11/site-packages (from smart-open<8.0.0,>=5.2.1->weasel<0.5.0,>=0.1.0->spacy) (1.17.3)\n",
      "Requirement already satisfied: mdurl~=0.1 in /User/.conda/envs/daniel311/lib/python3.11/site-packages (from markdown-it-py>=2.2.0->rich>=10.11.0->typer<1.0.0,>=0.3.0->spacy) (0.1.2)\n",
      "Downloading spacy-3.8.7-cp311-cp311-manylinux_2_17_x86_64.manylinux2014_x86_64.whl (33.0 MB)\n",
      "\u001b[2K   \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m33.0/33.0 MB\u001b[0m \u001b[31m30.0 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m00:01\u001b[0m00:01\u001b[0m\n",
      "\u001b[?25hDownloading catalogue-2.0.10-py3-none-any.whl (17 kB)\n",
      "Downloading cymem-2.0.11-cp311-cp311-manylinux_2_17_x86_64.manylinux2014_x86_64.whl (218 kB)\n",
      "\u001b[2K   \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m218.9/218.9 kB\u001b[0m \u001b[31m23.8 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m\n",
      "\u001b[?25hDownloading langcodes-3.5.0-py3-none-any.whl (182 kB)\n",
      "\u001b[2K   \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m183.0/183.0 kB\u001b[0m \u001b[31m17.8 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m\n",
      "\u001b[?25hDownloading murmurhash-1.0.13-cp311-cp311-manylinux_2_5_x86_64.manylinux1_x86_64.manylinux_2_17_x86_64.manylinux2014_x86_64.whl (123 kB)\n",
      "\u001b[2K   \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m123.4/123.4 kB\u001b[0m \u001b[31m15.4 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m\n",
      "\u001b[?25hDownloading preshed-3.0.10-cp311-cp311-manylinux_2_5_x86_64.manylinux1_x86_64.manylinux_2_17_x86_64.manylinux2014_x86_64.whl (842 kB)\n",
      "\u001b[2K   \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m842.9/842.9 kB\u001b[0m \u001b[31m6.5 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0ma \u001b[36m0:00:01\u001b[0mm\n",
      "\u001b[?25hDownloading spacy_legacy-3.0.12-py2.py3-none-any.whl (29 kB)\n",
      "Downloading spacy_loggers-1.0.5-py3-none-any.whl (22 kB)\n",
      "Downloading srsly-2.5.1-cp311-cp311-manylinux_2_17_x86_64.manylinux2014_x86_64.whl (1.1 MB)\n",
      "\u001b[2K   \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m1.1/1.1 MB\u001b[0m \u001b[31m4.9 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m00:01\u001b[0m00:01\u001b[0m\n",
      "\u001b[?25hDownloading thinc-8.3.6-cp311-cp311-manylinux_2_17_x86_64.manylinux2014_x86_64.whl (4.5 MB)\n",
      "\u001b[2K   \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m4.5/4.5 MB\u001b[0m \u001b[31m43.6 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m00:01\u001b[0m:00:01\u001b[0m\n",
      "\u001b[?25hDownloading numpy-2.3.3-cp311-cp311-manylinux_2_27_x86_64.manylinux_2_28_x86_64.whl (16.9 MB)\n",
      "\u001b[2K   \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m16.9/16.9 MB\u001b[0m \u001b[31m35.3 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m00:01\u001b[0m00:01\u001b[0m\n",
      "\u001b[?25hDownloading wasabi-1.1.3-py3-none-any.whl (27 kB)\n",
      "Downloading weasel-0.4.1-py3-none-any.whl (50 kB)\n",
      "\u001b[2K   \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m50.3/50.3 kB\u001b[0m \u001b[31m3.1 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m\n",
      "\u001b[?25hDownloading blis-1.3.0-cp311-cp311-manylinux_2_17_x86_64.manylinux2014_x86_64.whl (11.7 MB)\n",
      "\u001b[2K   \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m11.7/11.7 MB\u001b[0m \u001b[31m47.0 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m00:01\u001b[0m0:01\u001b[0mm\n",
      "\u001b[?25hDownloading cloudpathlib-0.22.0-py3-none-any.whl (61 kB)\n",
      "\u001b[2K   \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m61.5/61.5 kB\u001b[0m \u001b[31m7.0 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m\n",
      "\u001b[?25hDownloading confection-0.1.5-py3-none-any.whl (35 kB)\n",
      "Downloading language_data-1.3.0-py3-none-any.whl (5.4 MB)\n",
      "\u001b[2K   \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m5.4/5.4 MB\u001b[0m \u001b[31m52.3 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0ma \u001b[36m0:00:01\u001b[0mm\n",
      "\u001b[?25hDownloading smart_open-7.3.1-py3-none-any.whl (61 kB)\n",
      "\u001b[2K   \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m61.7/61.7 kB\u001b[0m \u001b[31m8.0 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m\n",
      "\u001b[?25hDownloading marisa_trie-1.3.1-cp311-cp311-manylinux_2_24_x86_64.manylinux_2_28_x86_64.whl (1.3 MB)\n",
      "\u001b[2K   \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m1.3/1.3 MB\u001b[0m \u001b[31m54.4 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m\n",
      "\u001b[?25hInstalling collected packages: cymem, wasabi, spacy-loggers, spacy-legacy, smart-open, numpy, murmurhash, marisa-trie, cloudpathlib, catalogue, srsly, preshed, language-data, blis, langcodes, confection, weasel, thinc, spacy\n",
      "  Attempting uninstall: numpy\n",
      "    Found existing installation: numpy 1.26.4\n",
      "    Uninstalling numpy-1.26.4:\n",
      "      Successfully uninstalled numpy-1.26.4\n",
      "\u001b[31mERROR: pip's dependency resolver does not currently take into account all the packages that are installed. This behaviour is the source of the following dependency conflicts.\n",
      "datasets 4.0.0 requires fsspec[http]<=2025.3.0,>=2023.1.0, but you have fsspec 2025.7.0 which is incompatible.\n",
      "mlrun 1.10.0rc26 requires numpy<1.27.0,>=1.26.4, but you have numpy 2.3.3 which is incompatible.\n",
      "pandas 2.1.4 requires numpy<2,>=1.23.2; python_version == \"3.11\", but you have numpy 2.3.3 which is incompatible.\n",
      "storey 1.10.13 requires numpy<1.27,>=1.16.5, but you have numpy 2.3.3 which is incompatible.\n",
      "unstructured 0.16.12 requires numpy<2, but you have numpy 2.3.3 which is incompatible.\u001b[0m\u001b[31m\n",
      "\u001b[0mSuccessfully installed blis-1.3.0 catalogue-2.0.10 cloudpathlib-0.22.0 confection-0.1.5 cymem-2.0.11 langcodes-3.5.0 language-data-1.3.0 marisa-trie-1.3.1 murmurhash-1.0.13 numpy-2.3.3 preshed-3.0.10 smart-open-7.3.1 spacy-3.8.7 spacy-legacy-3.0.12 spacy-loggers-1.0.5 srsly-2.5.1 thinc-8.3.6 wasabi-1.1.3 weasel-0.4.1\n",
      "Collecting en-core-web-sm==3.8.0\n",
      "  Downloading https://github.com/explosion/spacy-models/releases/download/en_core_web_sm-3.8.0/en_core_web_sm-3.8.0-py3-none-any.whl (12.8 MB)\n",
      "\u001b[2K     \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m12.8/12.8 MB\u001b[0m \u001b[31m43.6 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m00:01\u001b[0m00:01\u001b[0m\n",
      "\u001b[?25hInstalling collected packages: en-core-web-sm\n",
      "Successfully installed en-core-web-sm-3.8.0\n",
      "\u001b[38;5;2m✔ Download and installation successful\u001b[0m\n",
      "You can now load the package via spacy.load('en_core_web_sm')\n"
     ]
    }
   ],
   "source": [
    "# uncomment to install spacy requirements locally\n",
    "!pip install spacy\n",
    "!python -m spacy download en_core_web_sm"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**In this example**\n",
    "- [Create the pipeline](#create-the-pipeline)\n",
    "- [Test the pipeline locally](#test-the-pipeline-locally)\n",
    "- [Deploy to the cluster](#deploy-to-the-cluster)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Create the pipeline\n",
    "\n",
    "The pipeline consists of two functions: data-prep and NLP. Each one has different package dependencies.\n",
    "\n",
    "**Create a file with data-prep graph steps:**\n",
    "\n",
    "```{admonition} Note\n",
    "The `model`, `version` and `operation` can also be specified in the message body \n",
    "to support streaming protocols (e.g. Kafka).\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Overwriting data_prep.py\n"
     ]
    }
   ],
   "source": [
    "%%writefile data_prep.py\n",
    "import mlrun\n",
    "import json\n",
    "\n",
    "# load struct from a json file (event points to the url)\n",
    "def load_url(event):\n",
    "    url = event[\"url\"]\n",
    "    data = mlrun.get_object(url).decode(\"utf-8\")\n",
    "    return {\"url\": url, \"doc\": json.loads(data)}\n",
    "\n",
    "def to_paragraphs(event):\n",
    "    paragraphs = []\n",
    "    url = event[\"url\"]\n",
    "    for i, paragraph in enumerate(event[\"doc\"]):\n",
    "        paragraphs.append(\n",
    "            {\"url\": url, \"paragraph_id\": i, \"paragraph\": paragraph}\n",
    "        )\n",
    "    return paragraphs"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Create a file with NLP graph steps (use spacy):**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Overwriting nlp.py\n"
     ]
    }
   ],
   "source": [
    "%%writefile nlp.py\n",
    "import json\n",
    "import spacy\n",
    "\n",
    "def myprint(x):\n",
    "    print(x)\n",
    "    return x\n",
    "\n",
    "class ApplyNLP:\n",
    "    def __init__(self, context=None, spacy_dict=\"en_core_web_sm\"):\n",
    "\n",
    "        self.nlp = spacy.load(spacy_dict)\n",
    "\n",
    "    def do(self, paragraph: dict):\n",
    "        tokenized_paragraphs = []\n",
    "        if isinstance(paragraph, (str, bytes)):\n",
    "            paragraph = json.loads(paragraph)\n",
    "        tokenized = {\n",
    "            \"url\": paragraph[\"url\"],\n",
    "            \"paragraph_id\": paragraph[\"paragraph_id\"],\n",
    "            \"tokens\": self.nlp(paragraph[\"paragraph\"]),\n",
    "        }\n",
    "        tokenized_paragraphs.append(tokenized)\n",
    "\n",
    "        return tokenized_paragraphs\n",
    "\n",
    "def extract_entities(tokens):\n",
    "    paragraph_entities = []\n",
    "    for token in tokens:\n",
    "        entities = token[\"tokens\"].ents\n",
    "        for entity in entities:\n",
    "            paragraph_entities.append(\n",
    "                {\n",
    "                    \"url\": token[\"url\"],\n",
    "                    \"paragraph_id\": token[\"paragraph_id\"],\n",
    "                    \"entity\": entity.ents,\n",
    "                }\n",
    "            )\n",
    "    return paragraph_entities\n",
    "\n",
    "def enrich_entities(entities):\n",
    "    enriched_entities = []\n",
    "    for entity in entities:\n",
    "        enriched_entities.append(\n",
    "            {\n",
    "                \"url\": entity[\"url\"],\n",
    "                \"paragraph_id\": entity[\"paragraph_id\"],\n",
    "                \"entity_text\": entity[\"entity\"][0].text,\n",
    "                \"entity_start_char\": entity[\"entity\"][0].start_char,\n",
    "                \"entity_end_char\": entity[\"entity\"][0].end_char,\n",
    "                \"entity_label\": entity[\"entity\"][0].label_,\n",
    "            }\n",
    "        )\n",
    "    return enriched_entities"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Build and show the graph:**\n",
    "\n",
    "Create the master function (\"multi-func\") with the `data_prep.py` source and an async graph topology. \n",
    "Add a pipeline of steps made of custom python handlers, classes and built-in classes (like `storey.FlatMap`).\n",
    "\n",
    "The pipeline runs across two functions which are connected by a queue/stream (q1). Use the `function=` to specify which function runs the specified step.\n",
    "End the flow with writing to the output stream."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<mlrun.serving.states.QueueStep at 0x7f9d01234f10>"
      ]
     },
     "execution_count": 31,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# define a new real-time serving function (from code) with an async graph\n",
    "fn = project.set_function(\n",
    "    name=\"multi-func\", func=\"./data_prep.py\", kind=\"serving\", image=\"mlrun/mlrun\"\n",
    ")\n",
    "graph = fn.set_topology(\"flow\", engine=\"async\")\n",
    "\n",
    "# define the graph steps (DAG)\n",
    "graph.to(name=\"load_url\", handler=\"load_url\").to(\n",
    "    name=\"to_paragraphs\", handler=\"to_paragraphs\"\n",
    ").to(\"storey.FlatMap\", \"flatten_paragraphs\", _fn=\"(event)\").to(\n",
    "    \">>\", \"q1\", path=internal_stream\n",
    ").to(name=\"nlp\", class_name=\"ApplyNLP\", function=\"enrich\").to(\n",
    "    name=\"extract_entities\", handler=\"extract_entities\", function=\"enrich\"\n",
    ").to(name=\"enrich_entities\", handler=\"enrich_entities\", function=\"enrich\").to(\n",
    "    \"storey.FlatMap\", \"flatten_entities\", _fn=\"(event)\", function=\"enrich\"\n",
    ").to(name=\"printer\", handler=\"myprint\", function=\"enrich\").to(\n",
    "    \">>\", \"output_stream\", path=out_stream\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "image/svg+xml": [
       "<?xml version=\"1.0\" encoding=\"UTF-8\" standalone=\"no\"?>\n",
       "<!DOCTYPE svg PUBLIC \"-//W3C//DTD SVG 1.1//EN\"\n",
       " \"http://www.w3.org/Graphics/SVG/1.1/DTD/svg11.dtd\">\n",
       "<!-- Generated by graphviz version 2.43.0 (0)\n",
       " -->\n",
       "<!-- Title: mlrun&#45;flow Pages: 1 -->\n",
       "<svg width=\"1689pt\" height=\"44pt\"\n",
       " viewBox=\"0.00 0.00 1689.31 44.00\" xmlns=\"http://www.w3.org/2000/svg\" xmlns:xlink=\"http://www.w3.org/1999/xlink\">\n",
       "<g id=\"graph0\" class=\"graph\" transform=\"scale(1 1) rotate(0) translate(4 40)\">\n",
       "<title>mlrun&#45;flow</title>\n",
       "<polygon fill=\"white\" stroke=\"transparent\" points=\"-4,4 -4,-40 1685.31,-40 1685.31,4 -4,4\"/>\n",
       "<!-- _start -->\n",
       "<g id=\"node1\" class=\"node\">\n",
       "<title>_start</title>\n",
       "<polygon fill=\"lightgrey\" stroke=\"black\" points=\"38.55,-0.05 40.7,-0.15 42.83,-0.3 44.92,-0.49 46.98,-0.74 48.99,-1.03 50.95,-1.36 52.84,-1.75 54.66,-2.18 56.4,-2.65 58.06,-3.16 59.63,-3.71 61.11,-4.31 62.49,-4.94 63.76,-5.61 64.93,-6.31 65.99,-7.04 66.93,-7.8 67.77,-8.59 68.48,-9.41 69.09,-10.25 69.58,-11.11 69.95,-11.99 70.21,-12.89 70.36,-13.8 70.4,-14.72 70.33,-15.65 70.16,-16.59 69.89,-17.53 69.53,-18.47 69.07,-19.41 68.52,-20.35 67.89,-21.28 67.18,-22.2 66.4,-23.11 65.55,-24.01 64.63,-24.89 63.65,-25.75 62.62,-26.59 61.53,-27.41 60.4,-28.2 59.23,-28.96 58.02,-29.69 56.78,-30.39 55.5,-31.06 54.2,-31.69 52.88,-32.29 51.53,-32.84 50.17,-33.35 48.79,-33.82 47.4,-34.25 46,-34.64 44.59,-34.97 43.17,-35.26 41.75,-35.51 40.32,-35.7 38.89,-35.85 37.45,-35.95 36.02,-36 34.58,-36 33.15,-35.95 31.71,-35.85 30.28,-35.7 28.85,-35.51 27.43,-35.26 26.01,-34.97 24.6,-34.64 23.2,-34.25 21.81,-33.82 20.43,-33.35 19.07,-32.84 17.72,-32.29 16.4,-31.69 15.1,-31.06 13.82,-30.39 12.58,-29.69 11.37,-28.96 10.2,-28.2 9.07,-27.41 7.98,-26.59 6.95,-25.75 5.97,-24.89 5.05,-24.01 4.2,-23.11 3.42,-22.2 2.71,-21.28 2.08,-20.35 1.53,-19.41 1.07,-18.47 0.71,-17.53 0.44,-16.59 0.27,-15.65 0.2,-14.72 0.24,-13.8 0.39,-12.89 0.65,-11.99 1.02,-11.11 1.51,-10.25 2.11,-9.41 2.83,-8.59 3.66,-7.8 4.61,-7.04 5.67,-6.31 6.84,-5.61 8.11,-4.94 9.49,-4.31 10.97,-3.71 12.54,-3.16 14.2,-2.65 15.94,-2.18 17.76,-1.75 19.65,-1.36 21.61,-1.03 23.62,-0.74 25.68,-0.49 27.77,-0.3 29.9,-0.15 32.05,-0.05 34.22,0 36.38,0 38.55,-0.05\"/>\n",
       "<text text-anchor=\"middle\" x=\"35.3\" y=\"-14.3\" font-family=\"Times,serif\" font-size=\"14.00\">start</text>\n",
       "</g>\n",
       "<!-- load_url -->\n",
       "<g id=\"node2\" class=\"node\">\n",
       "<title>load_url</title>\n",
       "<ellipse fill=\"none\" stroke=\"black\" cx=\"154.69\" cy=\"-18\" rx=\"48.19\" ry=\"18\"/>\n",
       "<text text-anchor=\"middle\" x=\"154.69\" y=\"-14.3\" font-family=\"Times,serif\" font-size=\"14.00\">load_url</text>\n",
       "</g>\n",
       "<!-- _start&#45;&gt;load_url -->\n",
       "<g id=\"edge1\" class=\"edge\">\n",
       "<title>_start&#45;&gt;load_url</title>\n",
       "<path fill=\"none\" stroke=\"black\" d=\"M69.94,-18C78.21,-18 87.3,-18 96.34,-18\"/>\n",
       "<polygon fill=\"black\" stroke=\"black\" points=\"96.49,-21.5 106.49,-18 96.49,-14.5 96.49,-21.5\"/>\n",
       "</g>\n",
       "<!-- to_paragraphs -->\n",
       "<g id=\"node3\" class=\"node\">\n",
       "<title>to_paragraphs</title>\n",
       "<ellipse fill=\"none\" stroke=\"black\" cx=\"316.13\" cy=\"-18\" rx=\"77.19\" ry=\"18\"/>\n",
       "<text text-anchor=\"middle\" x=\"316.13\" y=\"-14.3\" font-family=\"Times,serif\" font-size=\"14.00\">to_paragraphs</text>\n",
       "</g>\n",
       "<!-- load_url&#45;&gt;to_paragraphs -->\n",
       "<g id=\"edge2\" class=\"edge\">\n",
       "<title>load_url&#45;&gt;to_paragraphs</title>\n",
       "<path fill=\"none\" stroke=\"black\" d=\"M203.1,-18C211.15,-18 219.73,-18 228.42,-18\"/>\n",
       "<polygon fill=\"black\" stroke=\"black\" points=\"228.64,-21.5 238.64,-18 228.64,-14.5 228.64,-21.5\"/>\n",
       "</g>\n",
       "<!-- flatten_paragraphs -->\n",
       "<g id=\"node4\" class=\"node\">\n",
       "<title>flatten_paragraphs</title>\n",
       "<ellipse fill=\"none\" stroke=\"black\" cx=\"528.27\" cy=\"-18\" rx=\"98.58\" ry=\"18\"/>\n",
       "<text text-anchor=\"middle\" x=\"528.27\" y=\"-14.3\" font-family=\"Times,serif\" font-size=\"14.00\">flatten_paragraphs</text>\n",
       "</g>\n",
       "<!-- to_paragraphs&#45;&gt;flatten_paragraphs -->\n",
       "<g id=\"edge3\" class=\"edge\">\n",
       "<title>to_paragraphs&#45;&gt;flatten_paragraphs</title>\n",
       "<path fill=\"none\" stroke=\"black\" d=\"M393.59,-18C401.94,-18 410.55,-18 419.19,-18\"/>\n",
       "<polygon fill=\"black\" stroke=\"black\" points=\"419.33,-21.5 429.33,-18 419.33,-14.5 419.33,-21.5\"/>\n",
       "</g>\n",
       "<!-- q1 -->\n",
       "<g id=\"node5\" class=\"node\">\n",
       "<title>q1</title>\n",
       "<polygon fill=\"none\" stroke=\"black\" points=\"705.06,-30 663.06,-30 663.06,-6 705.06,-6 717.06,-18 705.06,-30\"/>\n",
       "<text text-anchor=\"middle\" x=\"690.06\" y=\"-14.3\" font-family=\"Times,serif\" font-size=\"14.00\">q1</text>\n",
       "</g>\n",
       "<!-- flatten_paragraphs&#45;&gt;q1 -->\n",
       "<g id=\"edge4\" class=\"edge\">\n",
       "<title>flatten_paragraphs&#45;&gt;q1</title>\n",
       "<path fill=\"none\" stroke=\"black\" d=\"M627.28,-18C636.25,-18 644.9,-18 652.74,-18\"/>\n",
       "<polygon fill=\"black\" stroke=\"black\" points=\"652.89,-21.5 662.89,-18 652.89,-14.5 652.89,-21.5\"/>\n",
       "</g>\n",
       "<!-- nlp -->\n",
       "<g id=\"node6\" class=\"node\">\n",
       "<title>nlp</title>\n",
       "<ellipse fill=\"none\" stroke=\"black\" cx=\"780.06\" cy=\"-18\" rx=\"27\" ry=\"18\"/>\n",
       "<text text-anchor=\"middle\" x=\"780.06\" y=\"-14.3\" font-family=\"Times,serif\" font-size=\"14.00\">nlp</text>\n",
       "</g>\n",
       "<!-- q1&#45;&gt;nlp -->\n",
       "<g id=\"edge5\" class=\"edge\">\n",
       "<title>q1&#45;&gt;nlp</title>\n",
       "<path fill=\"none\" stroke=\"black\" d=\"M717.46,-18C725.45,-18 734.37,-18 742.88,-18\"/>\n",
       "<polygon fill=\"black\" stroke=\"black\" points=\"742.98,-21.5 752.98,-18 742.98,-14.5 742.98,-21.5\"/>\n",
       "</g>\n",
       "<!-- extract_entities -->\n",
       "<g id=\"node7\" class=\"node\">\n",
       "<title>extract_entities</title>\n",
       "<ellipse fill=\"none\" stroke=\"black\" cx=\"926.25\" cy=\"-18\" rx=\"83.39\" ry=\"18\"/>\n",
       "<text text-anchor=\"middle\" x=\"926.25\" y=\"-14.3\" font-family=\"Times,serif\" font-size=\"14.00\">extract_entities</text>\n",
       "</g>\n",
       "<!-- nlp&#45;&gt;extract_entities -->\n",
       "<g id=\"edge6\" class=\"edge\">\n",
       "<title>nlp&#45;&gt;extract_entities</title>\n",
       "<path fill=\"none\" stroke=\"black\" d=\"M807.17,-18C814.74,-18 823.47,-18 832.69,-18\"/>\n",
       "<polygon fill=\"black\" stroke=\"black\" points=\"832.85,-21.5 842.85,-18 832.85,-14.5 832.85,-21.5\"/>\n",
       "</g>\n",
       "<!-- enrich_entities -->\n",
       "<g id=\"node8\" class=\"node\">\n",
       "<title>enrich_entities</title>\n",
       "<ellipse fill=\"none\" stroke=\"black\" cx=\"1124.74\" cy=\"-18\" rx=\"79.09\" ry=\"18\"/>\n",
       "<text text-anchor=\"middle\" x=\"1124.74\" y=\"-14.3\" font-family=\"Times,serif\" font-size=\"14.00\">enrich_entities</text>\n",
       "</g>\n",
       "<!-- extract_entities&#45;&gt;enrich_entities -->\n",
       "<g id=\"edge7\" class=\"edge\">\n",
       "<title>extract_entities&#45;&gt;enrich_entities</title>\n",
       "<path fill=\"none\" stroke=\"black\" d=\"M1009.73,-18C1018.09,-18 1026.61,-18 1035.03,-18\"/>\n",
       "<polygon fill=\"black\" stroke=\"black\" points=\"1035.24,-21.5 1045.24,-18 1035.24,-14.5 1035.24,-21.5\"/>\n",
       "</g>\n",
       "<!-- flatten_entities -->\n",
       "<g id=\"node9\" class=\"node\">\n",
       "<title>flatten_entities</title>\n",
       "<ellipse fill=\"none\" stroke=\"black\" cx=\"1320.62\" cy=\"-18\" rx=\"80.69\" ry=\"18\"/>\n",
       "<text text-anchor=\"middle\" x=\"1320.62\" y=\"-14.3\" font-family=\"Times,serif\" font-size=\"14.00\">flatten_entities</text>\n",
       "</g>\n",
       "<!-- enrich_entities&#45;&gt;flatten_entities -->\n",
       "<g id=\"edge8\" class=\"edge\">\n",
       "<title>enrich_entities&#45;&gt;flatten_entities</title>\n",
       "<path fill=\"none\" stroke=\"black\" d=\"M1204.04,-18C1212.5,-18 1221.16,-18 1229.74,-18\"/>\n",
       "<polygon fill=\"black\" stroke=\"black\" points=\"1229.77,-21.5 1239.77,-18 1229.77,-14.5 1229.77,-21.5\"/>\n",
       "</g>\n",
       "<!-- printer -->\n",
       "<g id=\"node10\" class=\"node\">\n",
       "<title>printer</title>\n",
       "<ellipse fill=\"none\" stroke=\"black\" cx=\"1480.76\" cy=\"-18\" rx=\"43.59\" ry=\"18\"/>\n",
       "<text text-anchor=\"middle\" x=\"1480.76\" y=\"-14.3\" font-family=\"Times,serif\" font-size=\"14.00\">printer</text>\n",
       "</g>\n",
       "<!-- flatten_entities&#45;&gt;printer -->\n",
       "<g id=\"edge9\" class=\"edge\">\n",
       "<title>flatten_entities&#45;&gt;printer</title>\n",
       "<path fill=\"none\" stroke=\"black\" d=\"M1401.62,-18C1410.24,-18 1418.84,-18 1427.02,-18\"/>\n",
       "<polygon fill=\"black\" stroke=\"black\" points=\"1427.12,-21.5 1437.12,-18 1427.12,-14.5 1427.12,-21.5\"/>\n",
       "</g>\n",
       "<!-- output_stream -->\n",
       "<g id=\"node11\" class=\"node\">\n",
       "<title>output_stream</title>\n",
       "<polygon fill=\"none\" stroke=\"black\" points=\"1669.31,-30 1560.31,-30 1560.31,-6 1669.31,-6 1681.31,-18 1669.31,-30\"/>\n",
       "<text text-anchor=\"middle\" x=\"1620.81\" y=\"-14.3\" font-family=\"Times,serif\" font-size=\"14.00\">output_stream</text>\n",
       "</g>\n",
       "<!-- printer&#45;&gt;output_stream -->\n",
       "<g id=\"edge10\" class=\"edge\">\n",
       "<title>printer&#45;&gt;output_stream</title>\n",
       "<path fill=\"none\" stroke=\"black\" d=\"M1524.68,-18C1532.76,-18 1541.4,-18 1550.05,-18\"/>\n",
       "<polygon fill=\"black\" stroke=\"black\" points=\"1550.18,-21.5 1560.18,-18 1550.18,-14.5 1550.18,-21.5\"/>\n",
       "</g>\n",
       "</g>\n",
       "</svg>\n"
      ],
      "text/plain": [
       "<graphviz.graphs.Digraph at 0x7f9d0153f4d0>"
      ]
     },
     "execution_count": 36,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# specify the \"enrich\" child function, add extra package requirements\n",
    "child = fn.add_child_function(name=\"enrich\", url=\"./nlp.py\", image=\"mlrun/mlrun\")\n",
    "child.spec.build.commands = [\n",
    "    \"python -m pip install spacy\",\n",
    "    \"python -m spacy download en_core_web_sm\",\n",
    "]\n",
    "graph.plot(rankdir=\"LR\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Test the pipeline locally\n",
    "\n",
    "**Create an input file:**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Overwriting in.json\n"
     ]
    }
   ],
   "source": [
    "%%writefile in.json\n",
    "[\"Born and raised in Queens, New York City, Trump attended Fordham University for two years and received a bachelor's degree in economics from the Wharton School of the University of Pennsylvania. He became president of his father Fred Trump's real estate business in 1971, renamed it The Trump Organization, and expanded its operations to building or renovating skyscrapers, hotels, casinos, and golf courses. Trump later started various side ventures, mostly by licensing his name. Trump and his businesses have been involved in more than 4,000 state and federal legal actions, including six bankruptcies. He owned the Miss Universe brand of beauty pageants from 1996 to 2015, and produced and hosted the reality television series The Apprentice from 2004 to 2015.\", \n",
    " \"Trump's political positions have been described as populist, protectionist, isolationist, and nationalist. He entered the 2016 presidential race as a Republican and was elected in a surprise electoral college victory over Democratic nominee Hillary Clinton while losing the popular vote.[a] He became the oldest first-term U.S. president[b] and the first without prior military or government service. His election and policies have sparked numerous protests. Trump has made many false or misleading statements during his campaign and presidency. The statements have been documented by fact-checkers, and the media have widely described the phenomenon as unprecedented in American politics. Many of his comments and actions have been characterized as racially charged or racist.\"]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Create a mock server (simulator) and test:**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {},
   "outputs": [],
   "source": [
    "# tuggle verbosity if needed\n",
    "fn.verbose = True"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "{'url': 'in.json', 'paragraph_id': 0, 'entity_text': 'Queens', 'entity_start_char': 19, 'entity_end_char': 25, 'entity_label': 'GPE'}\n",
      "{'url': 'in.json', 'paragraph_id': 0, 'entity_text': 'New York City', 'entity_start_char': 27, 'entity_end_char': 40, 'entity_label': 'GPE'}\n",
      "{'url': 'in.json', 'paragraph_id': 0, 'entity_text': 'Trump', 'entity_start_char': 42, 'entity_end_char': 47, 'entity_label': 'ORG'}\n",
      "{'url': 'in.json', 'paragraph_id': 0, 'entity_text': 'Fordham University', 'entity_start_char': 57, 'entity_end_char': 75, 'entity_label': 'ORG'}\n",
      "{'url': 'in.json', 'paragraph_id': 0, 'entity_text': 'two years', 'entity_start_char': 80, 'entity_end_char': 89, 'entity_label': 'DATE'}\n",
      "{'url': 'in.json', 'paragraph_id': 0, 'entity_text': 'the Wharton School of the University of Pennsylvania', 'entity_start_char': 141, 'entity_end_char': 193, 'entity_label': 'ORG'}\n",
      "{'url': 'in.json', 'paragraph_id': 0, 'entity_text': \"Fred Trump's\", 'entity_start_char': 229, 'entity_end_char': 241, 'entity_label': 'PERSON'}\n",
      "{'url': 'in.json', 'paragraph_id': 0, 'entity_text': '1971', 'entity_start_char': 266, 'entity_end_char': 270, 'entity_label': 'DATE'}\n",
      "{'url': 'in.json', 'paragraph_id': 0, 'entity_text': 'The Trump Organization', 'entity_start_char': 283, 'entity_end_char': 305, 'entity_label': 'ORG'}\n",
      "{'url': 'in.json', 'paragraph_id': 0, 'entity_text': 'more than 4,000', 'entity_start_char': 529, 'entity_end_char': 544, 'entity_label': 'CARDINAL'}\n",
      "{'url': 'in.json', 'paragraph_id': 0, 'entity_text': 'six', 'entity_start_char': 588, 'entity_end_char': 591, 'entity_label': 'CARDINAL'}\n",
      "{'url': 'in.json', 'paragraph_id': 0, 'entity_text': 'Universe', 'entity_start_char': 624, 'entity_end_char': 632, 'entity_label': 'PERSON'}\n",
      "{'url': 'in.json', 'paragraph_id': 0, 'entity_text': '1996 to 2015', 'entity_start_char': 663, 'entity_end_char': 675, 'entity_label': 'DATE'}\n",
      "{'url': 'in.json', 'paragraph_id': 0, 'entity_text': 'The Apprentice', 'entity_start_char': 731, 'entity_end_char': 745, 'entity_label': 'WORK_OF_ART'}\n",
      "{'url': 'in.json', 'paragraph_id': 1, 'entity_text': 'Trump', 'entity_start_char': 0, 'entity_end_char': 5, 'entity_label': 'ORG'}\n",
      "{'url': 'in.json', 'paragraph_id': 1, 'entity_text': '2016', 'entity_start_char': 122, 'entity_end_char': 126, 'entity_label': 'DATE'}\n",
      "{'url': 'in.json', 'paragraph_id': 1, 'entity_text': 'Republican', 'entity_start_char': 150, 'entity_end_char': 160, 'entity_label': 'NORP'}\n",
      "{'url': 'in.json', 'paragraph_id': 1, 'entity_text': 'Democratic', 'entity_start_char': 222, 'entity_end_char': 232, 'entity_label': 'NORP'}\n",
      "{'url': 'in.json', 'paragraph_id': 1, 'entity_text': 'Hillary Clinton', 'entity_start_char': 241, 'entity_end_char': 256, 'entity_label': 'PERSON'}\n",
      "{'url': 'in.json', 'paragraph_id': 1, 'entity_text': 'first', 'entity_start_char': 312, 'entity_end_char': 317, 'entity_label': 'ORDINAL'}\n",
      "{'url': 'in.json', 'paragraph_id': 1, 'entity_text': 'U.S.', 'entity_start_char': 323, 'entity_end_char': 327, 'entity_label': 'GPE'}\n",
      "{'url': 'in.json', 'paragraph_id': 1, 'entity_text': 'first', 'entity_start_char': 349, 'entity_end_char': 354, 'entity_label': 'ORDINAL'}\n",
      "{'url': 'in.json', 'paragraph_id': 1, 'entity_text': 'Trump', 'entity_start_char': 459, 'entity_end_char': 464, 'entity_label': 'ORG'}\n",
      "{'url': 'in.json', 'paragraph_id': 1, 'entity_text': 'American', 'entity_start_char': 671, 'entity_end_char': 679, 'entity_label': 'NORP'}\n"
     ]
    }
   ],
   "source": [
    "from nlp import Apply\n",
    "\n",
    "# create a mock server (simulator), specify to simulate all the functions in the pipeline (\"*\")\n",
    "server = fn.to_mock_server(current_function=\"*\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# push a sample request into the pipeline and see the results print out (by the printer step)\n",
    "resp = server.test(body={\"url\": \"in.json\"})"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "server.wait_for_completion()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "server.wait_for_completion()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Deploy to the cluster"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# add credentials to the data/streams\n",
    "fn.apply(mlrun.platforms.v3io_cred())\n",
    "child.apply(mlrun.platforms.v3io_cred())\n",
    "\n",
    "# specify the error stream (to store exceptions from the functions)\n",
    "fn.spec.error_stream = err_stream\n",
    "\n",
    "# deploy as a set of serverless functions\n",
    "fn.deploy()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Listen on the output stream**\n",
    "\n",
    "You can use the SDK or CLI to listen on the output stream. Listening should be done in a separate console/notebook. Run:\n",
    "\n",
    "    mlrun watch-stream v3io:///users/admin/out-stream -j\n",
    "\n",
    "or use the SDK:\n",
    "```python\n",
    "from mlrun.platforms import watch_stream\n",
    "watch_stream(\"v3io:///users/admin/out-stream\", is_json=True)\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Test the live function:**\n",
    "```{Admonition} Note  \n",
    "The url must be a valid path to the input file.\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{'id': '79354e45-a158-405f-811c-976e9cf4ab5e'}"
      ]
     },
     "execution_count": 22,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "fn.invoke(\"\", body={\"url\": \"v3io:///users/admin/pipe/in.json\"})"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "daniel311",
   "language": "python",
   "name": "conda-env-.conda-daniel311-py"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.11.13"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}
